package com.zyx.concurrentdemo.mytest.c08;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author Yaxi.Zhang
 * @since 2021/10/19 09:22
 * desc: 自定义线程池
 */
@Slf4j(topic = "c.PoolDemo")
public class PoolDemo {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(1, 1, TimeUnit.SECONDS,1, (queue, task) -> {
            // 1. 死等
            // queue.put(task);
            // 2) 带有超时的等待
            queue.offer(task, 500, TimeUnit.MILLISECONDS);
            // 3) 让调用者放弃任务执行
            // log.debug("放弃{}", task);
            // 4) 让调用者抛出异常
            // throw new RuntimeException("任务执行失败 " + task);
            // 5) 让调用者自己执行任务
            // task.run();
        });

        for (int i = 0; i < 10; i++) {
            int j = i;
            threadPool.execute(() -> {
                log.debug("=====>>>>> execute");
                try {
                    Thread.sleep(3000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                log.debug("{}", j);
            });
        }

    }
}

@Slf4j(topic = "c.ThreadPool")
class ThreadPool {
    /**
     * 核心线程数
     */
    private final int corePoolsize;
    /**
     * 阻塞任务存放队列
     */
    private final BlockingQueue<Runnable> queue;

    /**
     * 线程集合
     */
    private final HashSet<Worker> workers = new HashSet<>();
    /**
     * 拒绝策略
     */
    private final RejectionPolicy<Runnable> rejectionPolicy;
    /**
     * 超时时间
     */
    private final long timeout;
    /**
     * 超时时间单位
     */
    private final TimeUnit timeUnit;

    public ThreadPool(int corePoolsize,
                      long timeout,
                      TimeUnit timeUnit,
                      int queueCapacity,
                      RejectionPolicy<Runnable> rejectionPolicy) {
        this.corePoolsize = corePoolsize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.queue = new BlockingQueue<>(queueCapacity);
        this.rejectionPolicy = rejectionPolicy;
    }

    /**
     * 执行任务
     */
    public void execute(Runnable task) {
        // 执行任务
        synchronized (workers) {
            if (workers.size() < corePoolsize) {
                // 工作线程小于线程池核心线程数, 直接放入workers中并执行
                Worker worker = new Worker(task);
                log.debug("ThreadPool 添加执行任务{}", task);
                workers.add(worker);
                worker.start();
            } else {
                // 工作线程不小于线程池核心线程数, 尝试将任务放入阻塞队列中
                // tryPut操作实际上由调用线程(main)执行!!! 可能会产生WAITING或者TIMED_WAITING造成阻塞
                queue.tryPut(rejectionPolicy, task);
            }
        }
    }

    class Worker extends Thread {

        private Runnable task;

        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            // 核心方法!!!!!!
            //      1.执行传入的任务
            //      2.执行传入的任务后从阻塞队列中取出任务, 直到任务队列中任务全部取出
            while (task != null || (task = queue.poll(timeout, timeUnit)) != null) {
                try {
                    log.debug("worker 正在执行...{}", task);
                    task.run();
                } catch (Exception exception) {
                    exception.printStackTrace();
                } finally {
                    task = null;
                }
            }
            // 执行完成任务后, 需要从线程集合中删除当前线程 !
            synchronized (workers) {
                log.debug("worker 被移除{}", this);
                workers.remove(this);
            }
        }

        @Override
        public int hashCode() {
            return super.hashCode();
        }

        @Override
        public boolean equals(Object obj) {
            return super.equals(obj);
        }
    }
}


@FunctionalInterface
interface RejectionPolicy<T> {
    /**
     * 拒绝策略
     *
     * @param queue 任务队列
     * @param task  传入的任务
     */
    void reject(BlockingQueue<T> queue, T task);
}

@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T> {
    /**
     * 任务队列(双端队列)
     */
    private final Deque<T> queue = new ArrayDeque<>();
    /**
     * 队列容量
     */
    private final int capacity;
    /**
     * 锁
     */
    private final ReentrantLock lock = new ReentrantLock();
    /**
     * 生产者条件变量
     */
    private final Condition fullSet = lock.newCondition();
    /**
     * 消费者条件变量
     */
    private final Condition emptySet = lock.newCondition();

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    /**
     * 阻塞生产
     */
    public void put(T task) {
        lock.lock();
        try {
            // 队列已满则进行等待
            while (queue.size() == capacity) {
                try {
                    log.debug("bockingqueue 等待加入任务队列 {} ...", task);
                    fullSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("bockingqueue 添加任务 {} ...", task);
            // 不满则添加
            queue.addLast(task);
            emptySet.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 带超时的阻塞生产
     */
    public boolean offer(T task, long timeout, TimeUnit timeUnit) {
        long nanos = timeUnit.toNanos(timeout);
        lock.lock();
        try {
            log.debug("task entering...");
            // 队列已满则进行等待
            while (queue.size() == capacity) {
                log.debug("entering blocking");
                try {
                    // 超过超时时间
                    if (nanos <= 0) {
                        log.debug("bockingqueue 加入任务{}超时...", task);
                        return false;
                    }
                    log.debug("bockingqueue 等待加入任务队列 {} ...", task);
                    nanos = fullSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("bockingqueue 添加任务 {} ...", task);
            // 不满则添加
            queue.addLast(task);
            emptySet.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞消费
     */
    public T take() {
        lock.lock();
        try {
            // 队列已空则进行等待
            while (queue.size() == 0) {
                try {
                    log.debug("bockingqueue 等待消费...");
                    emptySet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 不空则取出
            T task = queue.removeFirst();
            log.debug("bockingqueue 取出任务{}...", task);
            fullSet.signal();
            return task;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 带有超时时间的阻塞消费
     */
    public T poll(long timeout, TimeUnit timeUnit) {
        long nanos = timeUnit.toNanos(timeout);
        lock.lock();
        try {
            // 队列已空则进行等待
            while (queue.size() == 0) {
                try {
                    // 任务超时
                    if (nanos <= 0) {
                        log.debug("bockingqueue 取出任务超时...");
                        return null;
                    }
                    log.debug("bockingqueue 等待消费...");
                    nanos = emptySet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 不空则取出
            T task = queue.removeFirst();
            log.debug("bockingqueue 取出任务{}...", task);
            fullSet.signalAll();
            return task;
        } finally {
            lock.unlock();
        }
    }

    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 带有拒绝策略的阻塞生产
     */
    public void tryPut(RejectionPolicy<T> rejectionPolicy, T task) {
        lock.lock();
        try {
            // 队列已满则进行等待
            if (queue.size() == capacity) {
                rejectionPolicy.reject(this, task);
            } else {
                log.debug("bockingqueue 添加任务 {} ...", task);
                // 不满则添加
                queue.addLast(task);
                emptySet.signal();
            }
        } finally {
            lock.unlock();
        }
    }

}